#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <error.h>
#include <fcntl.h>
#include <assert.h>
#include <sys/time.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include <memory>
#include <string>
#include <map>
#include <list>
#include <vector>
#include <functional>
#include <atomic>

#include <mutex>
#include <thread>
#include <condition_variable>

using namespace std;

const int EPOLL_EVENT_COUNT = 10000;

#define now_us() \
({ \
	struct timeval tv; \
	gettimeofday(&tv, NULL); \
	(tv.tv_sec * 1000000 + tv.tv_usec); \
})

enum {
	EPOLL_WAIT = 0,
	EPOLL_WAIT_COST,
	EPOLL_WAIT_FD,
	EPOLL_PROCESS_COST,
};

enum {
	FD_SET = 0,
	FD_WAIT,
	FD_WAIT_COST,
	FD_PROCESS,
	FD_PROCESS_COST,
};

class Counter
{
public:
	Counter(int size) {
		for (int i = 0; i < size; i++) {
			auto ptr = shared_ptr<atomic<long>>(new atomic<long>(0));
			_values.push_back(ptr);
		}
	}

//	Counter(const vector<long>& v) {
//		for (auto& item : v) {
//			auto ptr = shared_ptr<atomic<long>>(new atomic<long>(item));
//			_values.push_back(ptr);
//		}
//	}

	Counter(const Counter& obj) {
		assert(_values.size() == obj._values.size());
		for (int i = 0; i < (int)_values.size(); i++) {
			auto v = obj._values[i]->load();
			_values[i]->store(v);
		}
	}

	Counter& operator=(const Counter& obj) {
		assert(_values.size() == obj._values.size());
		for (int i = 0; i < (int)_values.size(); i++) {
			auto v = obj._values[i]->load();
			_values[i]->store(v);
		}	
		return *this;
	}

	Counter& operator+=(const Counter& obj) {
		assert(_values.size() == obj._values.size());
		for (int i = 0; i < (int)_values.size(); i++) {
			_values[i]->store(_values[i]->load() + obj._values[i]->load());
		}
		return *this;
	}

	Counter operator+(const Counter& obj) {
		assert(_values.size() == obj._values.size());
		Counter result(_values.size());
		for (int i = 0; i < (int)_values.size(); i++) {
			result._values[i]->store(_values[i]->load() + obj._values[i]->load());
		}
		return result;
	}

	Counter operator-(const Counter& obj) {
		assert(_values.size() == obj._values.size());
		Counter result(_values.size());
		for (int i = 0; i < (int)_values.size(); i++) {
			result._values[i]->store(_values[i]->load() - obj._values[i]->load());
		}
		return result;
	}

	inline void add(int index, long v) {
		auto& ptr = *(_values[index].get());
		ptr += v;	
	}

	long get(int index) {
		return _values[index]->load();
	}

	vector<shared_ptr<atomic<long>>> _values;
};


class FdReactorEngine;
class FdHandleEngine;

FdReactorEngine* g_fd_reactor_engine = NULL;
FdHandleEngine*  g_fd_handle_engine  = NULL;

atomic<long> g_request_counter(0);
atomic<int>  g_connects(0);

int set_socket_unblock(int fd)
{
	int flag = fcntl(fd, F_GETFL, 0);
	if (flag == -1) {
		return -1;
	}
	return fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}

const int SLOT = 64;
atomic<long> g_counters[SLOT];

void add_request_counter(int fd)
{
	g_counters[fd % SLOT]++;
}

long get_request_counter()
{
	long total_counter = 0;
	for (int i = 0; i < SLOT; i++) {
		total_counter += g_counters[i].load();
	}
	return total_counter;
}

class FdContext
{
public:
    FdContext(int fd) {
        _fd = fd;
    }

    ~FdContext() {
        close(_fd);
    }

    virtual void handle_events(short events) = 0;

public:
    int _fd;
    int _events;
    string _send_buffer; 
    string _recv_buffer;
};

struct FdReactorEpoll
{
    int epoll_id;
};

class FdReactorEngine
{
public:
    FdReactorEngine(int count = 1) {
        for (int i = 0; i < count; i++) {
            FdReactorEpoll item;
            item.epoll_id = epoll_create(EPOLL_EVENT_COUNT);
            if (item.epoll_id == -1) {
                printf("epoll_create fail, errno:%d, error:%s\n", errno, strerror(errno));
                exit(1);
            }
            _fd_reactors.push_back(item); 
			_counters.push_back(new Counter(4));

            thread([this, i] {
				char name[1024];
				snprintf(name, sizeof(name), "fd_reactor_%d", i);
				pthread_setname_np(pthread_self(), name);
				run(i);
			}).detach();
			printf("create reactor_engine thread\n");
        }
    }

    bool set(FdContext* ptr, short events) {

        auto& fd_reactor = _fd_reactors[ptr->_fd % _fd_reactors.size()];

        struct epoll_event ev;
        ev.data.ptr = ptr;
        ev.events = events;

        int ret = epoll_ctl(fd_reactor.epoll_id, EPOLL_CTL_ADD, ptr->_fd, &ev);
        if (ret != 0) {
            printf(
                "epoll_ctl fail, epoll_id:%d, fd:%d, errno:%d, error:%s\n", 
                fd_reactor.epoll_id, 
                ptr->_fd, 
                errno, 
                strerror(errno)
            );
            return false;
        }

        return true;
    }

    virtual void handle_fd_events(FdContext* ptr, short events) {
        ;
    }

    void run(int index) {

        struct epoll_event events[EPOLL_EVENT_COUNT];

        int  epoll_id = _fd_reactors[index].epoll_id;
		auto &counter = _counters[index];

		while (1) {

			auto beg_us = now_us();
        	int ret = epoll_wait(epoll_id, events, EPOLL_EVENT_COUNT, -1);
        	if (ret == -1) {
        	    exit(0);
        	}
			auto end_us = now_us();

			counter->add(EPOLL_WAIT, 1);
			counter->add(EPOLL_WAIT_COST, end_us - beg_us);
			counter->add(EPOLL_WAIT_FD, ret);

			beg_us = now_us();	
        	for (int i = 0; i < ret; i++) {

				auto ptr = (FdContext*)events[i].data.ptr;
				int  ev  = events[i].events;

        	   	int ret = epoll_ctl(epoll_id, EPOLL_CTL_DEL, ptr->_fd, NULL);
        	   	if (ret == -1) {
        	   	    printf(
        	   	        "epoll_ctl fail, epoll_id:%d, op:del, fd:%d, errno:%d, error:%s",
        	   	        epoll_id,
        	   	        ptr->_fd,
        	   	        errno,
        	   	        strerror(errno)
        	   	    );
        	   	    exit(1);
        	   	}

        	    handle_fd_events(ptr, ev & (EPOLLIN | EPOLLOUT));
        	}
			end_us = now_us();	
			counter->add(EPOLL_PROCESS_COST, end_us - beg_us);
		}
    }

	Counter get_counter() {
		Counter total_counter(4);
		for (auto& item : _counters) {
			total_counter += *item;
		}
		return total_counter;
	}

private:
    vector<FdReactorEpoll> _fd_reactors;
	vector<Counter*> _counters;
};

struct FdEventInfo
{
	FdContext* ptr;
    short events;
};

class FdEventsList
{
public:
    FdEventsList() {
        _mutex = shared_ptr<mutex>(new mutex);
    }

    void append(const FdEventInfo& fd_event) {
        lock_guard<mutex> lock(*_mutex);
        _fd_events.push_back(fd_event);
    }

    void append(const list<FdEventInfo>& fd_events) {
        lock_guard<mutex> lock(*_mutex);
        _fd_events.insert(_fd_events.end(), fd_events.begin(), fd_events.end());
    }

    bool get(FdEventInfo& info) {
        lock_guard<mutex> lock(*_mutex);
        if (_fd_events.size() == 0) {
            return false;
        }
        info = _fd_events.front();
        _fd_events.pop_front();
        return true;
    }

    bool get(list<FdEventInfo>& fd_events) {
        lock_guard<mutex> lock(*_mutex);
        if (_fd_events.size() == 0) {
            return false;
        }
        swap(fd_events, _fd_events);
		return true;
    }

    size_t size() {
        lock_guard<mutex> lock(*_mutex);
        return _fd_events.size();
    }

private:
    shared_ptr<mutex> _mutex;
    list<FdEventInfo> _fd_events;
};

class FdHandleEngine
{
public:
    FdHandleEngine(int count) {
        _fd_events.resize(count);
        for (int i = 0; i < count; i++) {
			_counters.push_back(new Counter(5));
            thread([this, i] {
				char name[1024];
				snprintf(name, sizeof(name), "fd_handle_%d", i);
				pthread_setname_np(pthread_self(), name);
				run(i);
			}).detach();
			printf("create fd_engine thread\n");
        }
    }

    void set(const FdEventInfo& info) {
        bool notify = false;
        {
            lock_guard<mutex> lock(_mutex);
            auto& item = _fd_events[info.ptr->_fd % _fd_events.size()];
            if (item.size() == 0) {
                notify = true;
            }
            item.append(info);
        }
        if (notify) {
            _cv.notify_all();
        }
    }

    void run(int index) {

		auto& counter = _counters[index];

		while (1) {

        	list<FdEventInfo> lst_fd_events;

        	unique_lock<mutex> lock(_mutex);

			bool need_counter = false;
			if (_fd_events[index].size() == 0) {
				need_counter = true;
			}

			auto beg_us = now_us();
        	_cv.wait(
        	    lock, 
        	    [this, index] {
        	        return _fd_events[index].size() > 0;
        	    }
        	);
			auto end_us = now_us();
			if (need_counter) {
				counter->add(FD_WAIT, 1);
				counter->add(FD_WAIT_COST, end_us - beg_us);
			}
        	_fd_events[index].get(lst_fd_events);  
        	lock.unlock();

			counter->add(FD_PROCESS, 1);

			beg_us = now_us();
        	for (auto& item : lst_fd_events) {
				item.ptr->handle_events(item.events);
        	}
			end_us = now_us();
			counter->add(FD_PROCESS_COST, end_us - beg_us);
		}
    }

	Counter get_counter() {
		Counter total_counter(5);
		for (auto& item : _counters) {
			total_counter += *item;
		}
		return total_counter;
	}

private:
    vector<FdEventsList> _fd_events;
	vector<Counter*> _counters;

    mutex _mutex;
    condition_variable _cv;
};

class FdCtxConnector : public FdContext
{
public:
	FdCtxConnector(int fd) : FdContext(fd) {
		;
	}

	void handle_events(short events) {
		if (events & EPOLLIN) {
			handle_recv_event();
		}
		if (events & EPOLLOUT) {
			handle_send_event();
		}
	}

	void handle_recv_event() {

		while (1) {
			char buf[1024];
			int ret = recv(_fd, buf, sizeof(buf), 0);
			if (ret > 0) {
				_recv_buffer.append(buf, ret);
			} else if (ret == 0) {
				printf("%s|socket close, fd:%d\n", __FUNCTION__, _fd);
				exit(0);
			} else if (errno == EAGAIN || errno == EWOULDBLOCK) {
				break ;
			} else {
				printf(
					"%s|socket error, fd:%d, errno:%d, error:%s\n", 
					__FUNCTION__, 
					_fd, 
					errno, 
					strerror(errno)
				);
				exit(0);
			}
		}

		if (_recv_buffer.size() <= 4) {
			return ;
		}

	   	int size = ntohl(*((int*)_recv_buffer.c_str()));
		int total_size = size + sizeof(int);
		if ((int)_recv_buffer.size() < total_size) {
			return ;
		}

		_send_buffer = _recv_buffer;
		_recv_buffer.clear();

		g_request_counter++;
	//	add_request_counter(_fd);

	//	g_fd_reactor_engine->set(this, EPOLLOUT);
		reactor_engine_set(EPOLLOUT);
	}

	void handle_send_event() {

		while (1) {

			int send_size = _send_buffer.size();
			int ret = send(_fd, _send_buffer.c_str(), send_size, 0);
			if (ret > 0) {
				_send_buffer.erase(0, ret);	
				if (ret == send_size) {
				//	g_fd_reactor_engine->set(this, EPOLLIN);
					reactor_engine_set(EPOLLIN);
					break ;
				}
			} else if (errno == EAGAIN || errno == EWOULDBLOCK) {
			//	g_fd_reactor_engine->set(this, EPOLLOUT);
				reactor_engine_set(EPOLLOUT);
				break ;
			}
			else {
				printf(
					"%s|socket error, ret:%d, fd:%d, errno:%d, error:%s\n", 
					__FUNCTION__, 
					ret, 
					_fd, 
					errno, 
					strerror(errno)
				);
			}
		}
	}

	inline void reactor_engine_set(short events) {
		g_fd_reactor_engine->set(this, events);
	}
};

class FdCtxAcceptor : public FdContext
{
public:
	FdCtxAcceptor(int fd) : FdContext(fd) {
		;
	}

	void handle_events(short events) {
		struct sockaddr_in caddr;
		socklen_t len = sizeof(caddr);
		int c = accept(_fd, (struct sockaddr*)&caddr, &len);
		if (c == -1) {
			printf(
				"%s|accept error, fd:%d, errno:%d, error:%s\n", 
				__FUNCTION__, 
				_fd, 
				errno, 
				strerror(errno)
			);
			exit(0);
		}
		printf("%s|new connection, fd:%d\n", __FUNCTION__, c);

		int ret = set_socket_unblock(c);
		if (ret == -1) {
			printf(
				"%s|set new connection unblock fail, fd:%d, errno:%d, error:%s\n", 
				__FUNCTION__, 
				c, 
				errno, 
				strerror(errno)
			);
			exit(0);
		}

		g_connects++;
		if (!g_fd_reactor_engine->set(new FdCtxConnector(c), EPOLLIN)) {
			printf(
				"%s|reactor_engine fail, fd:%d, events:%d\n", 
				__FUNCTION__, 
				c, 
				EPOLLIN
			);
			exit(0);
		}

		if (!g_fd_reactor_engine->set(this, EPOLLIN)) {
			printf(
				"%s|reactor_engine fail, fd:%d, events:%d\n", 
				__FUNCTION__, 
				c, 
				EPOLLIN
			);
			exit(0);
		}
	}
};

class FdReactorEnginePlus : public FdReactorEngine
{
public:
    FdReactorEnginePlus(int count) : FdReactorEngine(count) {
        ;
    }

    void handle_fd_events(FdContext* ptr, short events) {
        g_fd_handle_engine->set({ptr, events});
    }
};

class FdReactorEnginePlus1 : public FdReactorEngine
{
public:
    FdReactorEnginePlus1(int count) : FdReactorEngine(count) {
        ;
    }

    void handle_fd_events(FdContext* ptr, short events) {
		ptr->handle_events(events);
    }
};

int main(int argc, char** argv)
{
	int port = 6666;
	int fd_reactor_engines = 1;

	if (argc >= 2) {
		fd_reactor_engines = atoi(argv[1]);
	}
	if (argc >= 3) {
		port = atoi(argv[2]);
	}
	printf("usage[0] >> fd_reactor_engines:%d\n", fd_reactor_engines);
	printf("usage[1] >> port:%d\n", port);

    g_fd_reactor_engine = new FdReactorEnginePlus1(fd_reactor_engines);

	sleep(1);

	int s = socket(AF_INET, SOCK_STREAM, 0);

	int opt = 1;
	int ret = setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (const void*)&opt, sizeof(opt));
	if (ret == -1) {
		printf("set socket reuse addr fail, fd:%d, errno:%d, error:%s\n", s, errno, strerror(errno));
		return -1;
	}
	
	struct sockaddr_in saddr;
	saddr.sin_family = AF_INET;
	saddr.sin_port   = htons(port);
	saddr.sin_addr.s_addr = INADDR_ANY;

	ret = bind(s, (struct sockaddr*)&saddr, sizeof(saddr));
	if (ret == -1) {
		printf("bind fail, fd:%d, errno:%d, error:%s\n", s, errno, strerror(errno));
		return -1;
	}

	ret = listen(s, 1000);
	if (ret == -1) {
		printf("listen fail, fd:%d, errno:%d, error:%s\n", s, errno, strerror(errno));
		return -1;
	}

	if (!g_fd_reactor_engine->set(new FdCtxAcceptor(s), EPOLLIN)) {
		printf("reactor_engine set fail, fd:%d, events:%d\n", s, EPOLLIN);
		return -1;
	}

	long last_request_counter = g_request_counter.load();
//	long last_request_counter = get_request_counter();
	auto last_reactor_counter = g_fd_reactor_engine->get_counter();

    while (1) {
        sleep(1);
		long now_request_counter = g_request_counter.load();
	//	long now_request_counter = get_request_counter();
		auto now_reactor_counter = g_fd_reactor_engine->get_counter();

		auto counter = now_reactor_counter - last_reactor_counter;
		
		printf(
			"[logic]      {connect:%d, request:%ld/s}\n", 
			g_connects.load(),
			now_request_counter - last_request_counter
		);

		printf(
			"[fd_reactor] {epoll_wait:%ld, epoll_wait_cost:%0.2fus, epoll_process_cost:%0.2fus, epoll_wait_fds:%0.2f, usage:%0.2f%%}\n", 
			counter.get(EPOLL_WAIT),
			(double)counter.get(EPOLL_WAIT_COST) / counter.get(EPOLL_WAIT),
			(double)counter.get(EPOLL_PROCESS_COST) / counter.get(EPOLL_WAIT),
			(double)counter.get(EPOLL_WAIT_FD) / counter.get(EPOLL_WAIT),
			((double)(counter.get(EPOLL_WAIT_COST) + counter.get(EPOLL_PROCESS_COST)) / fd_reactor_engines) / 10000
		);

		printf("\n");

		last_request_counter = now_request_counter;
		last_reactor_counter = now_reactor_counter;
    }
    return 0;
}

